//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// index_scan_executor.cpp
//
// Identification: src/execution/index_scan_executor.cpp
//
// Copyright (c) 2015-19, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//
#include "execution/executors/index_scan_executor.h"
#include <utility>
#include "catalog/catalog.h"
#include "common/macros.h"
#include "storage/index/b_plus_tree_index.h"
#include "storage/table/tuple.h"

namespace bustub {
IndexScanExecutor::IndexScanExecutor(ExecutorContext *exec_ctx, const IndexScanPlanNode *plan)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      tree_(GetTree(exec_ctx, plan)),
      iterator_(GetTree(exec_ctx, plan)->GetBeginIterator()) {}

auto inline IndexScanExecutor::GetTree(ExecutorContext *exec_ctx, const IndexScanPlanNode *plan)
    -> BPlusTreeIndexForTwoIntegerColumn * {
  return dynamic_cast<BPlusTreeIndexForTwoIntegerColumn *>(
      exec_ctx_->GetCatalog()->GetIndex(plan_->index_oid_)->index_.get());
}

void IndexScanExecutor::Init() {
  IndexInfo *index_info = GetExecutorContext()->GetCatalog()->GetIndex(plan_->index_oid_);
  table_info_ = GetExecutorContext()->GetCatalog()->GetTable(index_info->table_name_);
  iterator_ = GetTree(exec_ctx_, plan_)->GetBeginIterator();
}

auto IndexScanExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (iterator_ == GetTree(GetExecutorContext(), plan_)->GetEndIterator()) {
    return false;
  }
  *rid = (*iterator_).second;
  auto tuple_meta = table_info_->table_->GetTuple(*rid).first;
  while (tuple_meta.is_deleted_) {
    ++iterator_;
    if (iterator_.IsEnd()) {
      return false;
    }
    *rid = (*iterator_).second;
    tuple_meta = table_info_->table_->GetTuple(*rid).first;
  }
  *tuple = table_info_->table_->GetTuple(*rid).second;
  ++iterator_;
  return true;
}

}  // namespace bustub
